package api.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import utils.PropertiesReader;

import java.util.Properties;

/**
 * Sink kafka
 */
public class SinkTest1_kafka {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties props = new Properties();

        String kafkaServers = PropertiesReader.get("default.kafka.servers");
        String topicFrom = PropertiesReader.get("default.kafka.topic");
        String topicTo = "myTest2";
        props.setProperty("bootstrap.servers", kafkaServers);
        props.setProperty("group.id", "flink-consumer-SinkTest1");

        // 从Kafka中读取数据
        DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer(topicFrom, new SimpleStringSchema(), props));
        // 将空格转成逗号
        DataStream<String> dataStream = inputStream.map(line -> {
            return line.replace(" ", ",");
        });
        // 将数据写入Kafka
        dataStream.addSink( new FlinkKafkaProducer<String>(kafkaServers, topicTo, new SimpleStringSchema()));
        env.execute();
    }
}
